#include "server.h"

static int _read(struct worker_ctl *worker_ctl,
        struct ep_active_slot *active_slot)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *msg_in=active_slot->_u._read.msg;
    int res;

    msg_in[active_slot->_u._read.msg_size]=ZERO;

    if(active_slot->slot == worker_ctl->listener_slot)
    {/*{{{*/
        char *body;
        char *svc;
        char para_desc[100];
        PARA para;

        body=msg_in;
        if(!(svc=para_get_item(msg_in, &body)) ||
                (res=strlen(svc)) <= 0 || res > NAME_SIZE)
        {
            errlog("%s %s: svc(%s) invalid, impossible",
                    __FILE__, __func__, (svc)?svc:SPACES);
            return(-1);
        }
        if(strcmp(svc, "NEWSLOT") == 0)
        {/*{{{*/
            char *value;
            SOCKET new_sock;
            EP_SLOT new_slot;
            struct ep_slot_info slot_info;
            struct _client_info *client_info;

            sprintf(para_desc, "sock(number:%d)", SHORT_NUMBER_SIZE);
            para_load(body, para_desc, &para);
            para_ref_by_name(para, "sock", &value);
            new_sock=atoi(value);
            para_free(para);

            if((res=ep_slot_register(worker_ctl->ep, new_sock, IS_FALSE,
                            &new_slot)))
            {
                errlog("%s %s: worker(%d) ep_slot_register fail, res:%d",
                        __FILE__, __func__, worker_idx, res);
                sock_close(new_sock);
                return(-1);
            }
            ep_slot_set_wrap(new_slot);

            ep_get_slot_info(worker_ctl->ep, new_slot, &slot_info);
            client_info=&share_info.worker_info[worker_idx].
                client_info[slot_info.slot_idx];
            pthread_mutex_lock(&share_info.worker_info_lock); //lock
            client_info->ip=strdup(slot_info.sock_info.peer_ip);
            client_info->port=slot_info.sock_info.peer_port;
            client_info->loop=client_info->count=0;
            pthread_mutex_unlock(&share_info.worker_info_lock); //unlock
            worker_ctl->client_count[slot_info.slot_idx]=0;

            errlog("%s %s: worker(%d) new slot",
                    __FILE__, __func__, worker_idx);
        }/*}}}*/
        else
        {
            errlog("%s %s: svc(%s) not found, impossible",
                    __FILE__, __func__, svc);
            return(-1);
        }
    }/*}}}*/
    else
    {/*{{{*/
        extern int svc_call(struct worker_ctl *worker_ctl, char *msg_in);

        char *msg_out=worker_ctl->msg_out;
        int msg_out_size;

        if((res=svc_call(worker_ctl, msg_in)))
        {
            struct ep_slot_info slot_info;
            ep_get_slot_info(worker_ctl->ep, active_slot->slot,
                    &slot_info);
            errlog("%s %s: worker(%d) svc_call res(%d), "
                    "from(%s:%d)",
                    __FILE__, __func__, worker_idx, res,
                    slot_info.sock_info.peer_ip,
                    slot_info.sock_info.peer_port);
        }
        ++worker_ctl->client_count[active_slot->slot_idx];

        msg_out_size=strlen(msg_out);
        if((res=ep_write(worker_ctl->ep, active_slot->slot,
                        msg_out, &msg_out_size)) != 0)
        {
            errlog("%s %s: worker(%d) ep_write fail, res:%d",
                    __FILE__, __func__, worker_idx, res);
            return(-1);
        }
    }/*}}}*/

    return(0);
}/*}}}*/

static int _abort(struct worker_ctl *worker_ctl,
                        struct ep_active_slot *active_slot)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    struct _worker_info *worker_info;
    struct _client_info *client_info;
    int client_num;

    if(active_slot->slot == worker_ctl->listener_slot)
    {
        errlog("%s %s: worker(%d) revoked by listener, res:%d, "
                "impossible",
                __FILE__, __func__, worker_idx,
                active_slot->_u._abort.error);
        return(-1);
    }

    worker_info=&share_info.worker_info[worker_idx];
    client_info=&worker_info->client_info[active_slot->slot_idx];
    pthread_mutex_lock(&share_info.worker_info_lock);      // lock
    FREE(client_info->ip);
    client_num=--worker_info->client_num;
    pthread_mutex_unlock(&share_info.worker_info_lock);    // unlock
    worker_ctl->client_count[active_slot->slot_idx]=0;

    errlog("%s %s: worker(%d) client log off, total:%d "
            "revoke slot(sock:%d, idx:%d), res:%d",
            __FILE__, __func__, worker_idx, client_num,
            active_slot->_u._abort.sock,
            active_slot->slot_idx,
            active_slot->_u._abort.error);
    return(0);
}/*}}}*/

static int _timeout(struct worker_ctl *worker_ctl,
                        struct ep_active_slot *active_slot)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    struct _worker_info *worker_info;
    struct _client_info *client_info;
    int client_num;

    if(active_slot->slot == worker_ctl->listener_slot)
    {
        errlog("%s %s: worker(%d) listener timeout, res:%d, "
                "impossible",
                __FILE__, __func__, worker_idx,
                active_slot->_u._abort.error);
        return(-1);
    }

    worker_info=&share_info.worker_info[worker_idx];
    client_info=&worker_info->client_info[active_slot->slot_idx];
    pthread_mutex_lock(&share_info.worker_info_lock);      // lock
    FREE(client_info->ip);
    client_num=--worker_info->client_num;
    pthread_mutex_unlock(&share_info.worker_info_lock);    // unlock
    worker_ctl->client_count[active_slot->slot_idx]=0;

    errlog("%s %s: worker(%d) timeout, total:%d "
            "revoke slot(sock:%d, idx:%d), res:%d",
            __FILE__, __func__, worker_idx, client_num,
            active_slot->_u._abort.sock,
            active_slot->slot_idx,
            active_slot->_u._abort.error);
    return(0);
}/*}}}*/

void thrd_worker(struct worker_ctl *worker_ctl)
{/*{{{*/
#define UPDATE_SVC_INFO_INTERVAL            10
#define COUNT_NUM                           1000000
    int worker_idx=worker_ctl->worker_idx;
    struct ep_active_slot *active_list;
    time_t now;
    int res;

    while(1)
    {
        if((res=ep_poll(worker_ctl->ep, &active_list)) != 0)
        {
            errlog("%s: worker(%d) revoke listener, res:%d, "
                    "impossible",
                    __func__, worker_idx, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_READ:
                    _read(worker_ctl, active_list);
                    break;
                case EP_SLOT_ABORT:
                    _abort(worker_ctl, active_list);
                    break;
                case EP_SLOT_TIMEOUT:
                    _timeout(worker_ctl, active_list);
                    break;
                default:
                    errlog("%s: worker(%d) poll type(%s) invalid, "
                            "impossible",
                            __func__, worker_idx, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }

        // update client & svc info in share_info
        now=curr_timestamp(NULL);
        if(now - worker_ctl->last_call > UPDATE_SVC_INFO_INTERVAL)
        {
            int i;

            pthread_mutex_lock(&share_info.svc_info_lock); //lock
            for(i=0; i<share_info.svc_num; ++i)
            {
                if(worker_ctl->svc_count[i] > 0)
                {
                    struct _svc_info *svc_info=&share_info.svc_info[i];
                    svc_info->count+=worker_ctl->svc_count[i];
                    if(svc_info->count >= COUNT_NUM)
                    {
                        svc_info->loop+=svc_info->count/COUNT_NUM;
                        svc_info->count%=COUNT_NUM;
                    }
                    worker_ctl->svc_count[i]=0;
                }
            }
            pthread_mutex_unlock(&share_info.svc_info_lock); //unlock

            pthread_mutex_lock(&share_info.worker_info_lock); //lock
            for(i=0; i<EP_POLL_MAX_NUM; ++i)
            {
                if(worker_ctl->client_count[i] > 0)
                {
                    struct _client_info *client_info=
                        &share_info.worker_info[worker_idx].
                        client_info[i];
                    client_info->count+=worker_ctl->client_count[i];
                    if(client_info->count >= COUNT_NUM)
                    {
                        client_info->loop+=client_info->count/COUNT_NUM;
                        client_info->count%=COUNT_NUM;
                    }
                    worker_ctl->client_count[i]=0;
                }
            }
            pthread_mutex_unlock(&share_info.worker_info_lock); //unlock

            worker_ctl->last_call=now;
        }
    }
}/*}}}*/
